package com.suxin.kafka.consumer;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * @author Tang
 * @classname KafaKaConsumerExample
 * @description [  ]
 * @date 2021/7/20 14:17
 */
@Slf4j
@Component
public class KafaKaConsumerExample {

    @KafkaListener(
            id = "kafkaOneConsumer",
            topics = {"one-kafka-test"},
            groupId = "one-group"
    )
    public void kafkaOneConsumer(ConsumerRecord<String, String> record, Consumer<String, String> consumer) {
        log.info("kafkaOneConsumer:--->" + record.value() + ",offset--->" + record.offset());
    }


    /**
     * 批处理测试
     * @param records
     */
    @KafkaListener(
            id = "kafkaBatchConsumer",
            topics = {"batch-kafka-test"},
            groupId = "batch-group",
            containerFactory = "batchKafkaListenerContainerFactory"
    )
    public void kafkaBatchConsumer(List<ConsumerRecord<String, String>> records) {
        log.info("接收到的记录条数为:{}...",records.size());
        records.forEach(record -> {
            log.info("kafkaOneConsumer:--->" + record.value() + ",offset--->" + record.offset());
        });
    }

    /**
     * 并行消费能力  Consumer 线程不安全，，需要单独处理
     * @param record
     * @param consumer
     */
//    @KafkaListener(
//            id = "kafkaFourFiveConsumer",
//            topics = {"four-kafka-test","five-kafka-test"},
//            groupId = "four-five-group",
//            containerFactory = "concurrentListenerContainerFactory"
//    )
//    public void kafkaFourFiveConsumer(ConsumerRecord<String, String> record, Consumer<String, String> consumer) {
//        log.info("kafkaFourFiveConsumer:--->" + record.value() + ",offset--->" + record.offset());
//        try {
//            TimeUnit.SECONDS.sleep(2);
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        }
//    }

    /**
     * 手动commit offset
     * @param record
     * @param consumer
     */
//    @KafkaListener(
//            id = "kafkaOneConsumer",
//            topics = {"one-kafka-test"},
//            groupId = "one-group"
//    )
//    public void kafkaOneConsumer(ConsumerRecord<String, String> record, Consumer<String, String> consumer) {
//        log.info("kafkaOneConsumer:--->" + record.value() + ",offset--->" + record.offset());
//        consumer.commitSync();
//    }

    /**
     * 同时监听多个 topic,,多个分区
     * @param record
     */
//    @KafkaListener(
//            id = "kafkaOneConsumer1",
//            topics = {"one-kafka-test"},
//            groupId = "one-group",
//            topicPartitions = {
//                    @TopicPartition(topic = "one-kafka-test",partitions = "0")
//            }
//    )
//    public void kafkaOneConsumer1(ConsumerRecord<String,String> record) {
//        System.out.println("kafkaOneConsumer1:--->" + record.value());
//    }

    @KafkaListener(
            id = "kafkaTwoConsumer",
            topics = {"two-kafka-test"},
            groupId = "two-group",
            containerFactory = "kafkaListenerTwoContainerFactory")
    public void kafkaTwoConsumer(ConsumerRecord<String, String> record) {
        log.info("kafkaTwoConsumer:--->" + record.value());
    }


    /**
     * 演示 同一个组内 多分区消费
     * three-kafka-test：2个分区
     * 默认分区策略：Range
     */
//    @KafkaListener(
//            id = "kafkaThreeConsumer1",
//            topics = {"three-kafka-test"},
//            groupId = "three-group",
//            containerFactory = "kafkaListenerTwoContainerFactory")
//    public void kafkaThreeConsumer1(ConsumerRecord<String, String> record) {
//        System.out.println("kafkaThreeConsumer1:--->" + record.value());
//    }
//
//    @KafkaListener(
//            id = "kafkaThreeConsumer2",
//            topics = {"three-kafka-test"},
//            groupId = "three-group",
//            containerFactory = "kafkaListenerTwoContainerFactory")
//    public void kafkaThreeConsumer2(ConsumerRecord<String, String> record) {
//        System.out.println("kafkaThreeConsumer2:--->" + record.value());
//    }

}